
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

public class hbaseTest {

    private static final String NAME_SPACE = "vicchui";
    private static final String TABLE_NAME = "student";
    private static final String COLUMN_FAMILY1 = "info";
    private static final String COLUMN_FAMILY2 = "score";

    public static void createTable( ) throws IOException {
        Configuration config = HBaseConfiguration.create();
        config.set("zookeeper.znode.parent","/hbase"); //与 hbase-site-xml里面的配置信息 zookeeper.znode.parent 一致
        config.set("hbase.zookeeper.quorum","jikehadoop01,jikehadoop02,jikehadoop03");  //hbase 服务地址
        config.set("hbase.zookeeper.property.clientPort","2181"); //端口号
        try (Connection connection = ConnectionFactory.createConnection(config);
             Admin admin = connection.getAdmin()) {

             TableName tableName = TableName.valueOf(NAME_SPACE,TABLE_NAME);

             String[] colFamily = {COLUMN_FAMILY1, COLUMN_FAMILY2};

             boolean flag = admin.tableExists(tableName);
             if (flag) {
                 // table existed and delete it
                admin.disableTable(tableName);
                admin.deleteTable(tableName);
                System.out.println(tableName + " deleted.");
                admin.close();
                connection.close();
                System.exit(0);
             } else{
                //HTableDescriptor table = new HTableDescriptor(TableName.valueOf(TABLE_NAME));
                 // use hbase API 2.0.6 ; HTableDescriptor has been deprecated.
                 TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableName);
                //table.addFamily(new HColumnDescriptor(CF_DEFAULT).setCompactionCompressionType(Compression.Algorithm.NONE));
                // add column family to collection
                List<ColumnFamilyDescriptor> colFamDescList = new ArrayList<>();

                for (String str : colFamily) {
                    ColumnFamilyDescriptor familyDescriptor = ColumnFamilyDescriptorBuilder.newBuilder(str.getBytes()).build();
                    colFamDescList.add(familyDescriptor);
                }

                TableDescriptor tableDescriptor = tableDescriptorBuilder.setColumnFamilies(colFamDescList).build();
                System.out.println("Creating table.");
                admin.createTable(tableDescriptor);
            }
            admin.close();
            connection.close();
            System.out.println(" Create table Done.");
        } catch (IOException e) {

            e.printStackTrace();
        }
    }

    public static void addData(String rowkey, String value1, String value2, String value3, String value4) throws IOException {
        Configuration config = HBaseConfiguration.create();
        config.set("zookeeper.znode.parent","/hbase"); //与 hbase-site-xml里面的配置信息 zookeeper.znode.parent 一致
        config.set("hbase.zookeeper.quorum","jikehadoop01,jikehadoop02,jikehadoop03");  //hbase 服务地址
        config.set("hbase.zookeeper.property.clientPort","2181"); //端口号
        try (Connection connection = ConnectionFactory.createConnection(config)) {

            TableName tableName = TableName.valueOf(NAME_SPACE,TABLE_NAME);

            Table table = connection.getTable(tableName);

            Put put = new Put(rowkey.getBytes());
            put.addColumn(COLUMN_FAMILY1.getBytes(),"student_id".getBytes(),value1.getBytes());
            put.addColumn(COLUMN_FAMILY1.getBytes(),"class".getBytes(), value2.getBytes());
            put.addColumn(COLUMN_FAMILY2.getBytes(),"understanding".getBytes(),value3.getBytes());
            put.addColumn(COLUMN_FAMILY2.getBytes(),"programming".getBytes(),value4.getBytes());
            table.put(put);
            table.close();
            connection.close();
            System.out.println(" Adding data done.");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void getData(String rowkey) throws IOException{
        Configuration config = HBaseConfiguration.create();
        config.set("zookeeper.znode.parent","/hbase"); //与 hbase-site-xml里面的配置信息 zookeeper.znode.parent 一致
        config.set("hbase.zookeeper.quorum","jikehadoop01,jikehadoop02,jikehadoop03");  //hbase 服务地址
        config.set("hbase.zookeeper.property.clientPort","2181"); //端口号
        try (Connection connection = ConnectionFactory.createConnection(config)) {

            TableName tableName = TableName.valueOf(NAME_SPACE,TABLE_NAME);

            Table table = connection.getTable(tableName);
            Get get = new Get(rowkey.getBytes());
            Result result = table.get(get);

           List<Cell> listCells = result.listCells();
           for (Cell cell : listCells) {

               byte[] rowBytes = CellUtil.cloneRow(cell);
               byte[] familyBytes = CellUtil.cloneFamily(cell);
               byte[] qualifierBytes = CellUtil.cloneQualifier(cell);
               byte[] valueBytes = CellUtil.cloneValue(cell);

               String row = Bytes.toString(rowBytes);
               String family = Bytes.toString(familyBytes);
               String qualifier = Bytes.toString(qualifierBytes);
               String value = Bytes.toString(valueBytes);
               System.out.println("rowkey:" + row +" column family:"+family+" qualifier:"+qualifier+" value:"+value);
           }
            table.close();
            connection.close();
            System.out.println("get data Done");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void scan() {

        Configuration config = HBaseConfiguration.create();

        config.set("zookeeper.znode.parent","/hbase"); //与 hbase-site-xml里面的配置信息 zookeeper.znode.parent 一致
        config.set("hbase.zookeeper.quorum","jikehadoop01,jikehadoop02,jikehadoop03");  //hbase 服务地址
        config.set("hbase.zookeeper.property.clientPort","2181"); //端口号
        try (Connection connection = ConnectionFactory.createConnection(config)) {

            TableName tableName = TableName.valueOf(NAME_SPACE,TABLE_NAME);

            Table table = connection.getTable(tableName);

            Scan scan = new Scan();
            ResultScanner rscan = table.getScanner(scan);

            System.out.println("===================================================================");

            for (Result rs: rscan) {

                String rowkey = Bytes.toString(rs.getRow());
                System.out.println("row key :" + rowkey);
                Cell[] cells = rs.rawCells();
                for (Cell cell : cells) {
                    System.out.println(Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(),
                            cell.getFamilyLength())+"::"
                    +Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(),
                            cell.getQualifierLength())+"::"
                    +Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
                }
                System.out.println("===================================================================");

            }
            table.close();
            connection.close();
            System.out.println("get full table Done");
        } catch (IOException e) {

            e.printStackTrace();
        }

    }
    public static void delData(String  rowkey ) {

        Configuration config = HBaseConfiguration.create();

        config.set("zookeeper.znode.parent","/hbase"); //与 hbase-site-xml里面的配置信息 zookeeper.znode.parent 一致
        config.set("hbase.zookeeper.quorum","jikehadoop01,jikehadoop02,jikehadoop03");  //hbase 服务地址
        config.set("hbase.zookeeper.property.clientPort","2181"); //端口号
        try (Connection connection = ConnectionFactory.createConnection(config)) {

            TableName tableName = TableName.valueOf(NAME_SPACE,TABLE_NAME);
            Table table = connection.getTable(tableName);
            Delete delete = new Delete(Bytes.toBytes(rowkey));
            table.delete(delete);
            table.close();
            connection.close();
            System.out.println("Delete rowkey: "+rowkey+" Done.");
        } catch (IOException e) {

            e.printStackTrace();
        }





    }

    public static void updateData(String rowkey, String columnFamily, String columnName, String value) throws IOException {
        Configuration config = HBaseConfiguration.create();
        config.set("zookeeper.znode.parent","/hbase"); //与 hbase-site-xml里面的配置信息 zookeeper.znode.parent 一致
        config.set("hbase.zookeeper.quorum","jikehadoop01,jikehadoop02,jikehadoop03");  //hbase 服务地址
        config.set("hbase.zookeeper.property.clientPort","2181"); //端口号
        try (Connection connection = ConnectionFactory.createConnection(config)) {
            TableName tableName = TableName.valueOf(NAME_SPACE,TABLE_NAME);
            Table table = connection.getTable(tableName);
            Put put = new Put(rowkey.getBytes());
            put.addColumn(Bytes.toBytes(columnFamily),Bytes.toBytes(columnName), Bytes.toBytes(value));
            table.put(put);
            table.close();
            connection.close();
            System.out.println("Data updated: rowkey:"+rowkey+"column family:"+columnFamily+"column name:"+columnName+"value:"+ value);
        } catch (IOException e) {

            e.printStackTrace();
        }

    }




    public static void main(String[] args) throws IOException {

            // create table
            createTable();
            // add data
            //String rowkey, String value1, String value2, String value3, String value4
            addData("Tom", "20210000000001", "1", "75", "82");
            addData("Jerry", "20210000000002", "1", "85", "67");
            addData("Jack", "20210000000003", "2", "80", "80");
            addData("Rose", "20210000000004", "2", "60", "61");
            addData("Vic", "20210698010279", "2", "81", "70");

            // get one row data
            getData("Vic");
            // get whole table
            scan();
            // delete one row data
            delData("Tom");
            // get whole data
            //String rowkey, String columnFamily, String columnName, String value
            updateData("Vic", COLUMN_FAMILY2,"programming","80");
            // Verify data update
            getData("Vic");
            // add the delete data back
            addData("Tom", "20210000000001", "1", "75", "82");
            // get the result
            scan();

        }
}
